
package org.ssssssss.magicapi.mqtt;

import net.dreamlu.iot.mqtt.codec.ByteBufferUtil;
import net.dreamlu.iot.mqtt.codec.MqttQoS;
import net.dreamlu.iot.mqtt.core.client.IMqttClientMessageListener;
import net.dreamlu.iot.mqtt.core.client.MqttClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.Timer;
import java.util.TimerTask;

/**
 * 客户端测试
 *
 * @author 
 */
public class MqttClientStarter {
	private static final Logger logger = LoggerFactory.getLogger(MqttClientStarter.class);
	private static MqttClient client = null;
	
	// 连接服务器
	public static boolean connect(String ip,int port,String name,String pwd) {
		if(client!=null&&client.isConnected()) {
			return true;
		}
		else {
			client = MqttClient.create()
				.ip("127.0.0.1")
				.port(1883)
				.username("mica")
				.password("mica")
				.connectListener(new MqttClientConnectListener())
				// 同步连接，注意：连接会阻塞
				.connectSync();	
			if(client!=null) {
				return true;
			}
		}
		return false;
	}
	// 订阅主题
	public static void subscribe(String topicName) {
		client.subQos0(topicName, (topic, payload) -> {
			logger.info(topic + '\t' + ByteBufferUtil.toString(payload));
		});		
	}
	// 取消订阅主题
	public static void unSubscribe(String topicName) {
		client.unSubscribe(topicName);
	}
	// 发布消息
	public static boolean publish(String topic,String msg) {
		return client.publish(topic, msg.getBytes(StandardCharsets.UTF_8));
	}
	
	
	public static void main(String[] args) {
		// 初始化 mqtt 客户端
		MqttClient client = MqttClient.create()
			.ip("127.0.0.1")
			.port(1883)
			.username("mica")
			.password("mica")
//			如果包体过大，建议将此参数设置和 maxBytesInMessage 一样大
//			.readBufferSize(1024 * 10)
//			最大包体长度,如果包体过大需要设置此参数
//			.maxBytesInMessage(1024 * 10)
//			.version(MqttVersion.MQTT_5)
//			连接监听
			.connectListener(new MqttClientConnectListener())
			.willMessage(builder -> {
				builder.topic("/test/offline")
					.messageText("down")
					.retain(false)
					.qos(MqttQoS.AT_MOST_ONCE);    // 遗嘱消息
			})
			.connect();

		client.subQos0("/test/123", new IMqttClientMessageListener() {
			@Override
			public void onSubscribed(String topicFilter, MqttQoS mqttQoS) {
				// 订阅成功之后触发，可在此处做一些业务逻辑
				logger.info("topicFilter:{} MqttQoS:{} 订阅成功！！！", topicFilter, mqttQoS);
			}

			@Override
			public void onMessage(String topic, ByteBuffer payload) {
				logger.info(topic + '\t' + ByteBufferUtil.toString(payload));
			}
		});

		Timer timer = new Timer();
		timer.schedule(new TimerTask() {
			@Override
			public void run() {
				client.publish("/test/client", "mica最牛皮".getBytes(StandardCharsets.UTF_8));
			}
		}, 1000, 2000);
	}
}
